package cn.gyn.gynrpc.core.registry.zk;

import cn.gyn.gynrpc.core.api.RpcException;
import cn.gyn.gynrpc.core.api.RegistryCenter;
import cn.gyn.gynrpc.core.meta.InstanceMeta;
import cn.gyn.gynrpc.core.meta.ServiceMeta;
import cn.gyn.gynrpc.core.registry.ChangedListener;
import cn.gyn.gynrpc.core.registry.Events;
import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Value;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
public class ZkRegistryCenter implements RegistryCenter {
    @Value("${gynrpc.zk.server:localhost:2181}")
    private String servers;

    @Value("${gynrpc.zk.root:gynrpc}")
    private String root;

    private CuratorFramework client = null;

    private List<TreeCache> caches = new ArrayList<>();

    private boolean running = false;

    @Override
    public void start() {
        if (running) {
            log.info(" ===> zk client has started to server[" + servers + "/" + root + "], ignored.");
            return;
        }
        //ExponentialBackoffRetry是指数补偿重试策略，重试时间成倍增加，第一次1秒，第二次2秒，第三次4秒重试，随着重传次数的增加，延迟的程度也会指数增长。说的通俗点，每次重试的时间间隔都是上一次的两倍。
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .retryPolicy(retryPolicy)
                .connectString(servers)
                .namespace(root)
                .build();
        //日志放真正执行方法之前，可以看出问题出在哪
        log.info(" ===> zk client starting to server [" + servers + "/" + root + "].");
        client.start();
    }

    @Override
    public synchronized void stop() {
        if (!running) {
            log.info(" ===> zk client isn't running to server[" + servers + "/" + root + "], ignored.");
            return;
        }
        log.info(" ===> zk tree cache closed.");
        caches.forEach(TreeCache::close);
        log.info(" ===> zk client stopped.");
        client.close();
    }

    /**
     * 注册方法 ：最终会将namespace:gynrpc注册到zk的根节点
     * zk里的节点一般分为两种：
     * 1、持久化节点；（service服务可以创建为持久化节点）
     * 2、临时节点（instance比如8081、8082这些实例可以创建为临时节点）
     * @param service 服务 :描述为全限定名
     * @param instance 当前节点实例 ：描述为字符串
     */
    @Override
    public void register(ServiceMeta service, InstanceMeta instance) {
        String servicePath = "/" + service.toPath();
        try {
            //如果不存在服务，创建持久化服务
            if (client.checkExists().forPath(servicePath) == null) {
                //forPath方法里的第二个变量没有任何意义，随便写一个标识:比如"service".getBytes(),或者传入服务元数据的parameters的Json,传递参数用
                client.create().withMode(CreateMode.PERSISTENT).forPath(servicePath, service.toMetas().getBytes());
            }
            //创建provider实例的临时节点
            String instancePath = servicePath + "/" + instance.toPath();
            log.info(" === > register to zk: " + instancePath);
            //代码含义：https://v.flomoapp.com/mine/?memo_id=MTE4NTcxODQz
            client.create().withMode(CreateMode.EPHEMERAL).forPath(instancePath, instance.toMetas().getBytes());
        } catch (Exception ex) {
            throw new RpcException(ex);
        }
    }

    @Override
    public void unRegister(ServiceMeta service, InstanceMeta instance) {
        String servicePath = "/" + service.toPath();
        try {
            //如果不存在服务，直接返回
            if (client.checkExists().forPath(servicePath) == null) {
                return;
            }
            //安静删除掉provider实例的临时节点，报错也不进行提示
            String instancePath = servicePath + "/" + instance.toPath();
            log.info(" === > unregister from zk: " + instance);
            client.delete().quietly().forPath(instancePath);
        } catch (Exception ex) {
            throw new RpcException(ex);
        }
    }

    @Override
    public List<InstanceMeta> fetchAll(ServiceMeta service) {
        String servicePath = "/" + service.toPath();
        try {
            //获取所有子节点
            List<String> nodes = client.getChildren().forPath(servicePath);
            log.info(" === > fetchAll from zk: " + servicePath);
            return mapInstances(nodes, servicePath);
        } catch (Exception ex) {
            throw new RpcException(ex);
        }
    }

    @NotNull
    //private static List<InstanceMeta> mapInstances(List<String> nodes, String servicePath) {
    private List<InstanceMeta> mapInstances(List<String> nodes, String servicePath) {
        return nodes.stream().map(x -> {
            String[] node = x.split("_");
            InstanceMeta instance = InstanceMeta.http(node[0], Integer.valueOf(node[1]));
            System.out.println("instance: " + instance.toUrl());

            //循环获取节点的metas信息
            String nodePath = servicePath + "/" + x;
            byte[] bytes;
            try {
                //client是类里的一个属性，因此方法不能是静态的，否则无法访问
                bytes = client.getData().forPath(nodePath);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }

            //new String(bytes)
            //HashMap params = JSON.parseObject(new String(bytes), HashMap.class);
            Map<String,Object> params = JSON.parseObject(new String(bytes));
            params.forEach((k, v) -> {
                System.out.println(k + "->" + v);
                instance.getParameters().put(k, v == null ? null : v.toString());
            });
            return instance;
        }).collect(Collectors.toList());
    }

    /**
     * 订阅provider变化并更新
     * 对于变动和请求之间的时间差有服务请求导致的失败，要通过后续的 高可用、重试、容错 来进行处理
     * @param service
     * @param listener
     */
    @SneakyThrows
    @Override
    public void subscribe(ServiceMeta service, ChangedListener listener) {
        //订阅也是zk约定的方法
        final TreeCache cache = TreeCache.newBuilder(client, "/" + service.toPath())
                .setCacheData(true)
                //最多两层？
                .setMaxDepth(2)
                .build();
        cache.getListenable().addListener(
                (curator, event) -> {
                    synchronized (ZkRegistryCenter.class){
                        if(running){
                            //有任何节点变动这里会执行
                            log.info("zk subscribe event : " + event);
                            //获取当前最新的服务节点，封装到Events数据结构里，通过执行listener里的fire方法给传出来
                            List<InstanceMeta> nodes = fetchAll(service);
                            listener.fire(new Events(nodes));
                        }
                    }
                }
        );
        cache.start();
        caches.add(cache);
    }
}
